package com.xzx.flink.streamapi.source;

import com.xzx.flink.bean.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * @author xinzhixuan
 * @version V1.0.0
 * @date 2021/4/20
 **/
public class SourceFromCollection {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<SensorReading> source = env.fromCollection(Arrays.asList(
                new SensorReading("sensor_1", 1L, 30.1),
                new SensorReading("sensor_2", 2L, 31.1),
                new SensorReading("sensor_3", 3L, 32.2),
                new SensorReading("sensor_4", 4L, 33.1)
        ));

        source.print("collection");// 该方法里面可以传一个sink的唯一标识

        DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 2, 3, 45, 56);
        integerDataStreamSource.print("int");

        env.execute(SourceFromCollection.class.getName());// 该方法里面可以传一个jobName
    }
}
